package com.zz.util;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.util.Properties;

public class SparkMySQLToHiveToKafkaUrl {

    public static void main(String[] args) {


        // 1. 初始化SparkSession
        SparkConf conf = new SparkConf()
                .setAppName("SparkMySQLToHiveToKafka")
                .setMaster("local[*]"); // 生产环境去掉setMaster或使用集群管理器

        // 1. 配置SparkSession连接Hive（使用MySQL作为metastore）
        SparkSession spark = SparkSession.builder()
                .config(conf) // 或 yarn
                .config("hive.metastore.uris", "thrift://hadoop101:9083")  // Hive Metastore 地址
                .config("hive.server2.proxy.user", "root")  // 指定代理用户
                .config("hive.server2.authentication", "NOSASL")  // 禁用认证（测试用）
                .enableHiveSupport()  // 启用 Hive 支持
                .getOrCreate();

        spark.sql("SHOW DATABASES").show();
        spark.sql("USE finder");


        try {
            // 2. 从MySQL读取配置数据
            Dataset<Row> mysqlConfigDF = readFromMySQL(spark);

            // 假设mysqlConfigDF包含表名和查询条件等信息
            // 这里我们假设每条记录包含一个表名和一个where条件
            mysqlConfigDF.foreach(row -> {
                int tableName = row.getInt(0);
                String whereCondition = row.getString(2);
                System.out.println(whereCondition);

                // 3. 拼接SQL查询Hive数据
                String hiveQuery = "SELECT * FROM  finder.json_data_table limit 10";

                // 4. 执行Hive查询
                Dataset<Row> hiveDataDF = spark.sql(hiveQuery);
                hiveDataDF.show(5);

//                hiveDataDF.foreach((ForeachFunction<Row>) x->
//                        System.out.println(x));

                // 5. 将数据写入Kafka
//                writeToKafka(hiveDataDF);
            });

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭SparkSession
            spark.stop();
        }
    }

    /**
     * 从MySQL读取配置数据
     */
    private static Dataset<Row> readFromMySQL(SparkSession spark) {
        // MySQL连接配置
        String jdbcUrl = "jdbc:mysql://hadoop101:3306/finder";
        String table = "finder_sql";
        Properties connectionProperties = new Properties();
        connectionProperties.put("user", "root");
        connectionProperties.put("password", "123456");
        connectionProperties.put("driver", "com.mysql.jdbc.Driver");

        // 读取MySQL数据
        return spark.read()
                .jdbc(jdbcUrl, table, connectionProperties);
    }

    /**
     * 将数据写入Kafka
     */
    private static void writeToKafka(Dataset<Row> dataDF) {
        // Kafka配置
        String kafkaBootstrapServers = "hadoop101:9092,hadoop102:9092,hadoop103";
        String topic = "finder";

        // 将DataFrame转换为JSON格式字符串
        Dataset<String> jsonDF = dataDF.toJSON();

        // 写入Kafka
        jsonDF.write()
                .format("kafka")
                .option("kafka.bootstrap.servers", kafkaBootstrapServers)
                .option("topic", topic)
                .save();
    }
}